package com.study.app

import com.alibaba.fastjson.serializer.SerializeConfig
import com.alibaba.fastjson.{JSON, JSONObject}
import com.study.bean.PageLog
import com.study.util.MyKafkaUtils
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @Description:
 * @Author: LiuQun
 * @Date: 2022/5/17 12:39
 */
/**
 * 日志数据的消费分流
 * 1.准备实时处理环境 StreamingContext
 *
 * 2.从Kafka中消费数据
 *
 * 3.处理数据
 *    3.1 转换数据结构
 * 专用结构  Bean
 * 通用结构  Map  JsonObject
 *    3.2 分流
 *
 * 4. 写成到DWD层
 */
object OdsBaseLog {
  def main(args: Array[String]): Unit = {
    //1.准备实时环境
    val sparkConf: SparkConf = new SparkConf().setAppName("ods_base_log_app").setMaster("local[4]")
    //采集周期
    val ssc = new StreamingContext(sparkConf, Seconds(5))

    //2.从Kafka中消费数据
    val topicName: String = "ODS_BASE_LOG_1018" //对应生成器配置中的主题名
    val groupId: String = "ODS_BASE_LOG_GROUP_1018"
    val kafkaStream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtils.getKafkaDStream(ssc, topicName, groupId)

    //3.处理数据
    val jsonObjDStream: DStream[JSONObject] = kafkaStream.map(
      consumerRecord => {
        //获取ConsumerRecord中的value，就是日志数据
        val log: String = consumerRecord.value()
        //转成Json对象
        val jsonObject: JSONObject = JSON.parseObject(log)
        //返回
        jsonObject
      }
    )
    // jsonObjDStream.print(1000)

    //3.2分流
    // 日志数据：
    //    页面访问数据
    //        公共字段
    //        页面数据
    //        曝光数据
    //        事件数据
    //        错误数据
    //    启动数据
    //        公共字段
    //        启动数据
    //        错误数据

    val DWD_PAGE_LOG_TOPIC: String = "DWD_PAGE_LOG_TOPIC_1018"
    val DWD_PAGE_DISPLAY_TOPIC: String = "DWD_PAGE_DISPLAY_TOPIC_1018"
    val DWD_PAGE_ACTION_TOPIC: String = "DWD_PAGE_ACTION_TOPIC_1018"
    val DWD_START_LOG_TOPIC: String = "DWD_START_LOG_TOPIC_1018"
    val DWD_ERROR_LOG_TOPIC: String = "DWD_ERROR_LOG_TOPIC_1018"

    //分流规则：
    //  错误数据：不做任何的拆分，只要包含错误字段，直接将整条数据发送到对应的topic
    //  页面数据：拆分成页面访问，曝光，事件分别发送到对应的topic
    //  启动试数据：发动到对应的topic
    jsonObjDStream.foreachRDD(rdd => {
      rdd.foreach(jsonObj => {
        //分流过程
        val errObj = jsonObj.getJSONObject("err")
        if (errObj != null) {
          //将错误数据发送到DWD_ERROR_LOG_TOPIC
          MyKafkaUtils.send(DWD_ERROR_LOG_TOPIC, jsonObj.toJSONString)
        } else {
          //提取公共字段
          val commonObj: JSONObject = jsonObj.getJSONObject("common")
          val ar: String = commonObj.getString("ar")
          val uid: String = commonObj.getString("uid")
          val os: String = commonObj.getString("os")
          val ch: String = commonObj.getString("ch")
          val isNew: String = commonObj.getString("is_new")
          val md: String = commonObj.getString("md")
          val mid: String = commonObj.getString("mid")
          val vc: String = commonObj.getString("vc")
          val ba: String = commonObj.getString("ba")
          //提取时间错
          val ts: Long = jsonObj.getLong("ts")
          //页面数据
          val pageObj: JSONObject = jsonObj.getJSONObject("page")
          if (pageObj != null){
            //提取page字段
            val pageId: String = pageObj.getString("page_id")
            val pageItem: String = pageObj.getString("item")
            val pageItemType: String = pageObj.getString("item_type")
            val duringTime: Long = pageObj.getLong("during_time")
            val lastPageId: String = pageObj.getString("last_page_id")
            val sourceType: String = pageObj.getString("source_type")

            //封装PageLog
            var pageLog: PageLog = PageLog(mid, uid, ar, ch, isNew, md, os, vc, ba, pageId, lastPageId, pageItem, pageItemType, duringTime, sourceType, ts)
            //发送到DWD_PAGE_LOG_TOPIC
            MyKafkaUtils.send(DWD_PAGE_LOG_TOPIC,JSON.toJSONString(pageLog,new SerializeConfig(true)))
            // TODO: p18 

          }
          //启动数据
        }
      })
    })

    ssc.start()
    ssc.awaitTermination()


  }
}
